package cn.itcast.nio.c2;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import static cn.itcast.nio.c1.ByteBufferUtil.debugAll;

/**
 * 对selectorServer的单线程的多路复用优化
 * 是多线程版本的 目的就是提高cpu的利用率 加快对客户端的响应 本质就是提高效率
 *
 * 之前的一个selector掌管着客户端服务端的channel 又要负责服务端的连接调度也要负责客户端的读写调度
 * 所以现在搞多了一个线程 专门用来处理读写的channel的调度
 *
 * BIO 阻塞IO  NIO新IO非阻塞IO  AIO异步IO(调用读/写方法 然后留下一个回调方法以后数据来了让另个线程走这个回调方法把读到数据给到我)
 *
 *
 * BIO阻塞IO  NIO非诸塞IO 她们都是全双工通信(读的同时也可以写) 全双工通讯就是双方可以同时通信 半双工就是只能交替通讯
 * 但是stream是单向(读/写)的  channel是双向(读写都可以)的
 *
 * 同步(当前线程卡在这里等待返回结果且当前线程接收到返回结果才往下走)  异步(当前线程不等待返回结果直接往下执行，当有返回结果了则另一个线程通知当前线程返回结果是什么)
 * 为什么要用异步这么理解 病人看病需要20分钟 那么医生就只能从头到尾自己去处理这个问题 其他的病人每次都得等20分钟 假如我们把看病分成挂号 看病 缴费 取药这几个步骤
 * 那么每个步骤估计5分钟 那么每个病人就不用等这么久了 所以异步提高效率的本质是因为可以提高吞吐量(单位时间内的处理请求个数),但是因为多个步骤之间需要协调，所以单个病人完全看完病的总时长可能 > 20分钟
 * 所以做异步的时候，只有进行合理的任务拆分才是提高效率的关键
 *
 *
 * IO模型详解 我们Java程序是不能直接获取网络的数据的,都是调用操作系统去获取然后操作系统返回给java程序
 * 我们java程序从操作系统拿到数据有两步  1.等待数据(网络数据给到操作系统)  2.复制数据(操作系统从网卡中读取到数据复制给java程序)
 *
 * 阻塞IO : 在等待数据阶段的时候Java程序就阻塞了 直到网络数据给到操作系统 操作系统再把数据复制给到java程序 阻塞才完成才继续往下走
 * 所以如果客户端一直没反应就一直卡住 做一件事的时候做不了另一件事   所以阻塞IO是同步的
 *
 * 非阻塞IO:在等待数据阶段java程序是不会卡住的 (每次问操作系统有没有数据鸭有的话就阻塞拿数据没有我就再问你有没有数据鸭)
 * 只有当操作系统接受完数据要复制给java程序的时候 java程序才会阻塞接收数据然后程序往下走 (等待不阻塞 复制阻塞)
 * 所以客户端就算没反应 java程序一样会往下走，但当客户端发送数据操作系统也接受完要复制给java程序的时候就会阻塞一下
 * 非阻塞IO也是同步的
 *
 *
 * 多路复用：可以理解为阻塞IO的调度版,阻塞IO会有弊端,参考我们写的代码,当前的线程建立连接发送数据后想再发一次数据但是因为另外线程连接没建立阻塞了
 * 所以导致当前线程的数据就一直没办法发送了,而selector调度者就是把等待数据跟发送数据根据事件拆分开了,等待数据的事件自己去阻塞 发送/接收数据的事件也自己去阻塞
 * 可以通过不同的key去区分处理不同的事件，这样另一个线程就算没建立连接也不影响当前线程继续发数据
 * 阻塞IO则为遇到事件的阻塞  selector(多路复用)则为把它拆分为很多个事件 根据不同的事件分别自己去阻塞，不要影响其他事件操作
 * 多路复用也是同步的
 *
 *
 * 异步就不可能阻塞，如果都有阻塞了怎么可能叫异步呢亲 所以异步都是非阻塞的
 *
 */
@Slf4j
public class MultiThreadServer {

    public static void main(String[] args) throws IOException {
        //我们以后再用多线程的时候一定要给线程带上名字，不然以后在排查多个线程的问题的时候不好定位
        Thread.currentThread().setName("boss");
        ServerSocketChannel ssc = ServerSocketChannel.open(); //建立服务端的channel
        ssc.bind(new InetSocketAddress(8080));  //服务端绑定8080
        ssc.configureBlocking(false); //设置服务端为非阻塞访问
        Selector boss = Selector.open(); //建立selector
        SelectionKey bossKey = ssc.register(boss, 0, null); //绑定selector跟服务端channel
        bossKey.interestOps(SelectionKey.OP_ACCEPT);  //关注连接事件

        //我们一般搞跟cpu核数一样多的worker来运行 我们这里模拟一下
//        Worker worker  = new Worker("worker-0");
        Worker[] workers = new Worker[2];
        //这个方法可以得到当前电脑能用的线程核心数
//        Worker[] workers = new Worker[Runtime.getRuntime().availableProcessors()];
        for (int i =0 ; i<workers.length; i++ ){
            workers[i] = new Worker("worker-"+i);
        }
        AtomicInteger index = new AtomicInteger(); //原子性Integer默认是0
        while (true){
            boss.select(); //无事件阻塞 有事件异步放入selectedKeys()的列表集合
            Iterator<SelectionKey> iter = boss.selectedKeys().iterator();
            while (iter.hasNext()){
                SelectionKey key = iter.next();
                iter.remove();
                if(key.isAcceptable()){ //如果是接收事件 就肯定是服务端的key
                    SocketChannel sc = ssc.accept(); //得到客户端的channel
                    sc.configureBlocking(false); //设置客户端为非阻塞
                    log.debug("connected...{}",sc.getRemoteAddress());
                    //把这里的channel 给到worker里面的selector 让worker里面的selector来执行读写操作
                    //这里的worker是个静态的内部类 所以是可以直接访问到私有的成员变量的
                    log.debug("before register...{}",sc.getRemoteAddress());
                    //这就能轮询 第一次 0 第二次 1 第三次又是0  index.getAndIncrement()就是i++
                    workers[index.getAndIncrement() % workers.length].register(sc);
                    //下面这行方法有问题的，因为这里跟上面的worker共用一个selector对象 但是他们是2个线程
                    //selector对象已经在上面的worker中且线程已经start了所以里面的select()会让线程阻塞住
                    //阻塞住后就会导致下面这行代码(这个线程)一直执行不了了
                    //我们可以考虑把register方法拿到离下面这行代码近一点 但是也只是一个概率学问题而已只是概率稍微大了点 不能根本解决问题

                    //解决思路：我们可以把select()跟下面这个register()代码放在通过线程中，这样我们就可以自己决定执行顺序了
                    //sc.register(worker.selector,SelectionKey.OP_READ,null);
                    log.debug("after register...{}",sc.getRemoteAddress());
                }
            }
        }
    }

   static class Worker implements Runnable{
        private Thread thread;
        private Selector selector;
        private String name;
        // volatile关键字可以保证内存可见性
        // 内存可见性，即当线程A对volatile变量的修改后，其他线程获取的volatile变量都是最新的。
        // 参考 https://blog.csdn.net/weixin_35691102/article/details/114256780
        private volatile boolean start = false; //默认表示还未初始化
        private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>(); //作为两个线程之间传递数据的通道

        public Worker(String name){
            this.name = name;
        }

        //初始化线程 初始化selector
        // 要想在两个线程之间传递数据 就可以考虑用一个队列作为数据传输的通道
        public void register(SocketChannel sc) throws IOException {
            if(!start){
                thread = new Thread(this,name);
                selector = Selector.open();
                thread.start();
                start = true;
            }
            //但是这行代码本质还是原本的线程执行的 不是worker-0线程执行的 我们得放在run()中去
            //sc.register(selector, SelectionKey.OP_READ, null);   我们netty中就也是类似这样做的所以多理解一点
            queue.add(()->{  //放入到队列中去 然后在worker-0线程中再取出来执行
                try {
                    sc.register(selector, SelectionKey.OP_READ, null);
                } catch (ClosedChannelException e) {
                    e.printStackTrace();
                }
            });
            selector.wakeup();  //把selector唤醒让她别阻塞了，所以下面的selector.select()会被唤醒
        }

        //检测读写事件
        @Override
        public void run() {
            while (true){
                try {
                    selector.select();
                    //从队列中取出任务 因为是队列所以poll也会移除掉
                    Runnable task = queue.poll();
                    if(task!=null){
                        task.run(); //相当于执行了sc.register(selector, SelectionKey.OP_READ, null);
                    }
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        //这里的闭包粘包(数据长度问题) 客户端正常/非正常断开连接也会发送一次可读事件的处理等这里我们不关注
                        if (key.isReadable()) {
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            SocketChannel channel = (SocketChannel) key.channel();
                            log.debug("read...{}",channel.getRemoteAddress());
                            channel.read(buffer);
                            buffer.flip();
                            debugAll(buffer);
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}
